package cn.zwx.common.rocket.consumer.service.impl;

import cn.zwx.base.enums.ResponseCodeEnum;
import cn.zwx.base.exception.BizWrapException;
import cn.zwx.common.rocket.consumer.bo.ConsumerListenerBO;
import cn.zwx.common.rocket.consumer.bo.SubscribeBO;
import cn.zwx.common.rocket.consumer.enums.ReturnCodeEnum;
import cn.zwx.common.rocket.consumer.listener.ConsumerListener;
import cn.zwx.common.rocket.consumer.service.MqConsumerService;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;

/**
 * @author zhangwenxue
 */
@Service
public class MqConsumerServiceImpl implements MqConsumerService {

    private static final Logger logger = LoggerFactory.getLogger(MqConsumerServiceImpl.class);

    @Resource
    private RocketMQProperties rocketMqProperties;

    @Override
    public void subscribeForDefault(List<SubscribeBO> boList, ConsumerListener consumerListener) throws BizWrapException {
        // 初始化 Consumer
        DefaultMQPushConsumer defaultMqPushConsumer = this.extractedSubscribe(boList);
        if (Objects.isNull(defaultMqPushConsumer)){
            return;
        }
        // 处理消息消费
        defaultMqPushConsumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            for (MessageExt messageExt : list) {
                final ConsumerListenerBO consumerListenerBO = new ConsumerListenerBO(new String(messageExt.getBody()), messageExt.getMsgId(), messageExt.getKeys(),
                        messageExt.getTopic(), messageExt.getTags());
               try {
                   final String code = consumerListener.apply(consumerListenerBO);
                   if (ReturnCodeEnum.SUCCESS.getCode().equals(code)){
                       logger.info("消费:{},tag:{},消息key:{},消息体:{}成功..",messageExt.getTopic(),messageExt.getTags(),messageExt.getKeys(),new String(messageExt.getBody()));
                       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                   }else {
                       logger.info(String.format("%s消费者开始接收主题%s下,消息key:%s,消费失败", defaultMqPushConsumer.getConsumerGroup() , messageExt.getTopic(),messageExt.getKeys()));
                       return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                   }
               }catch (Exception e){
                   logger.error("{}消费者开始接收主题{}下消息key:{}失败原因:{}",defaultMqPushConsumer.getConsumerGroup() , messageExt.getTopic(),messageExt.getKeys(),e.getCause(),e);
                   return ConsumeConcurrentlyStatus.RECONSUME_LATER;
               }

            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        try {
            defaultMqPushConsumer.start();
        } catch (MQClientException e) {
            throw new BizWrapException(ResponseCodeEnum.CONSUMER_START_MQ_ERROR);
        }

    }




    @Override
    public void subscribeForOrder(List<SubscribeBO> boList, ConsumerListener consumerListener) throws BizWrapException {
        DefaultMQPushConsumer defaultMqPushConsumer = this.extractedSubscribe(boList);
        if (Objects.isNull(defaultMqPushConsumer)){
            return;
        }
        // 顺序消息 消费组 必须跟非顺序消息分开 不然存在订阅关系不一致问题
        defaultMqPushConsumer.setConsumerGroup(rocketMqProperties.getConsumer().getGroup()+"-order");
        // 处理消息消费
        defaultMqPushConsumer.registerMessageListener((MessageListenerOrderly) (list, consumeConcurrentlyContext) -> {
            for (MessageExt messageExt : list) {
                final ConsumerListenerBO consumerListenerBO = new ConsumerListenerBO(new String(messageExt.getBody()), messageExt.getMsgId(), messageExt.getKeys(),
                        messageExt.getTopic(), messageExt.getTags());
                try {
                    final String code = consumerListener.apply(consumerListenerBO);
                    if (ReturnCodeEnum.SUCCESS.getCode().equals(code)){
                        logger.info("消费:{},tag:{},消息key:{},消息体:{}成功..",messageExt.getTopic(),messageExt.getTags(),messageExt.getKeys(),new String(messageExt.getBody()));
                        return ConsumeOrderlyStatus.SUCCESS;
                    }else {
                        logger.info(String.format("%s消费者开始接收主题%s下,消息key:%s,消费失败", defaultMqPushConsumer.getConsumerGroup() , messageExt.getTopic(),messageExt.getKeys()));
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                }catch (Exception e){
                    logger.error("{}消费者开始接收主题{}下消息key:{}失败原因:{}",defaultMqPushConsumer.getConsumerGroup() , messageExt.getTopic(),messageExt.getKeys(),e.getCause(),e);
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }

            }
            return ConsumeOrderlyStatus.SUCCESS;
        });
        try {
            defaultMqPushConsumer.start();
        } catch (MQClientException e) {
            throw new BizWrapException(ResponseCodeEnum.CONSUMER_START_MQ_ERROR);
        }

    }

    private DefaultMQPushConsumer extractedSubscribe(List<SubscribeBO> boList) throws BizWrapException {
        RocketMQProperties.Consumer propertiesConsumer = rocketMqProperties.getConsumer();
        DefaultMQPushConsumer defaultMqPushConsumer;
        if (ObjectUtils.isEmpty(propertiesConsumer.getAccessKey()) || ObjectUtils.isEmpty(propertiesConsumer.getSecretKey())){
            defaultMqPushConsumer = new DefaultMQPushConsumer(propertiesConsumer.getGroup(),Boolean.TRUE);
        }else {
            AclClientRPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(propertiesConsumer.getAccessKey(), propertiesConsumer.getSecretKey()));
            defaultMqPushConsumer = new DefaultMQPushConsumer(propertiesConsumer.getGroup(),rpcHook,
                    new AllocateMessageQueueAveragely(),Boolean.TRUE,null);
        }
        defaultMqPushConsumer.setNamesrvAddr(rocketMqProperties.getNameServer());
        defaultMqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        if (CollectionUtils.isEmpty(boList)){
            return null;
        }
        // 批量订阅消息
        for (SubscribeBO subscribeBO : boList) {
            try {
                defaultMqPushConsumer.subscribe(subscribeBO.getTopic(), StringUtils.isBlank(subscribeBO.getTag())?"*":subscribeBO.getTag());
            } catch (MQClientException e) {
                throw new BizWrapException(ResponseCodeEnum.SUBSCRIBE_MQ_ERROR);
            }
        }
        return defaultMqPushConsumer;
    }
}
